/*
 * Copyright (c) 2020 Fraunhofer FOKUS and others. All rights reserved.
 *
 * See the NOTICE file(s) distributed with this work for additional
 * information regarding copyright ownership.
 *
 * This program and the accompanying materials are made available under the
 * terms of the Eclipse Public License 2.0 which is available at
 * http://www.eclipse.org/legal/epl-2.0
 *
 * SPDX-License-Identifier: EPL-2.0
 *
 * Contact: mosaic@fokus.fraunhofer.de
 */


package org.eclipse.mosaic.fed.mqtt.ambassador;

import org.apache.commons.lang3.StringUtils;
import org.eclipse.mosaic.fed.mqtt.callback.DefaultCallback;
import org.eclipse.mosaic.fed.mqtt.config.CMqtt;
import org.eclipse.mosaic.fed.mqtt.config.CVehicleAdd;
import org.eclipse.mosaic.fed.mqtt.constants.MqttTopic;
import org.eclipse.mosaic.fed.mqtt.entity.StoreParameter;
import org.eclipse.mosaic.fed.mqtt.message.AbstractMqttMessage;
import org.eclipse.mosaic.fed.mqtt.message.ObuMessage;
import org.eclipse.mosaic.fed.mqtt.protocol.ObuProtocol;
import org.eclipse.mosaic.interactions.mqtt.MqttPublish;
import org.eclipse.mosaic.interactions.traffic.VehicleUpdates;
import org.eclipse.mosaic.interactions.vehicle.VehicleFederateAssignment;
import org.eclipse.mosaic.lib.geo.MutableGeoPoint;
import org.eclipse.mosaic.lib.objects.vehicle.VehicleData;
import org.eclipse.mosaic.lib.util.JsonUtils;
import org.eclipse.mosaic.lib.util.NameGenerator;
import org.eclipse.mosaic.lib.util.objects.ObjectInstantiation;
import org.eclipse.mosaic.rti.api.*;
import org.eclipse.mosaic.rti.api.parameters.AmbassadorParameter;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * Implementation of a {@link AbstractFederateAmbassador} for the Mqtt
 * It is used to publish message to cloud and to receive message from it
 */
public class MqttAmbassador extends AbstractFederateAmbassador {

    private static final Map<String, StoreStrategy> strategyMap = new HashMap<>(10);
    /**
     * Vehicle id map
     * Key: [Sim world vehicle id]  Value: [Real world vehicle id]
     */
    private static final Map<String, String> vehicleIdMap = new ConcurrentHashMap<>(10);

    /**
     * task queue - thread safe
     */
    private final Queue<MqttPublish> queue = new ConcurrentLinkedQueue<>();

    private final boolean pubDirect = Boolean.FALSE;

    private static MqttClient pub;
    private static MqttClient sub;

    private static CMqtt mqtt;

    public MqttAmbassador(AmbassadorParameter ambassadorParameter) {
        super(ambassadorParameter);
        try {
            // read the Mqtt configuration file
            mqtt = new ObjectInstantiation<>(CMqtt.class, log)
                    .readFile(ambassadorParameter.configuration);
        } catch (InstantiationException e) {
            log.error("Configuration object could not be instantiated: ", e);
        }
    }

    @Override
    protected void processTimeAdvanceGrant(long time) throws InternalFederateException {
        super.processTimeAdvanceGrant(time);

        // todo:checkout here if thread is over-creating
        if (!pubDirect) {
            MqttProcess process = new MqttProcess();
            Thread mqttThread = new Thread(process);
            mqttThread.start();
        }
    }


    @Override
    public void initialize(long startTime, long endTime) throws InternalFederateException {
        super.initialize(startTime, endTime);

        connectToFederate("remote", null, null);

        for (MqttTopic value : MqttTopic.values()) {
            subscribe(value.getTopic());
        }

        try {
            rti.requestAdvanceTime(startTime, 0, (byte) 1);
        } catch (IllegalValueException e) {
            log.error("Error during advanceTime request", e);
            throw new InternalFederateException(e);
        }
    }

    @Override
    public void connectToFederate(String host, InputStream in, InputStream err) throws InternalFederateException {
        connectToFederate(mqtt.getHost(), mqtt.getPort());
    }


    @Override
    public void connectToFederate(String host, int port) {
        String serverURI = "tcp://" + host + ":" + port;

        String pubId = "MOSAIC_PUB" + (int) (Math.random() * 100000000);;
        String subId = "MOSAIC_SUB" + (int) (Math.random() * 100000000);;
        try {
            pub = new MqttClient(serverURI, pubId, new MemoryPersistence());
            sub = new MqttClient(serverURI, subId, new MemoryPersistence());
        } catch (MqttException e) {
            throw new RuntimeException("Can not initialize mqtt client.");
        }

        MqttConnectOptions options = mqttConnectOptions(mqtt.getUsername(), mqtt.getPassword());

        String callbackRef = mqtt.getCallback();
        MqttCallback callback = new DefaultCallback(this);
        if (StringUtils.isNotBlank(callbackRef)) {
            try {
                Class<?> clazz = Class.forName(callbackRef, true, this.getClass().getClassLoader());
                callback = (MqttCallback) clazz.getConstructor(MqttAmbassador.class).newInstance(this);
            } catch (Exception e) {
                throw new RuntimeException("Can not get the callback from config.", e);
            }
        }

        pub.setCallback(callback);
        sub.setCallback(callback);
        try {
            pub.connect(options);
            sub.connect(options);
        } catch (MqttException e) {
            throw new RuntimeException("Mqtt connect error", e);
        }
    }

    /**
     * MQTT connect options
     */
    private MqttConnectOptions mqttConnectOptions(String userName, String passWord) {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(userName);
        options.setPassword(passWord.toCharArray());
        // default 30 seconds
        options.setConnectionTimeout(10);
        // default false
        options.setAutomaticReconnect(true);
        // default true
        options.setCleanSession(true);
        // default 60
        // options.setKeepAliveInterval(20);
        return options;
    }

    /**
     * This method processes the interaction.
     *
     * @param interaction The interaction that can be processed.
     */
    @Override
    public synchronized void processInteraction(Interaction interaction) {
        // processed directly...
        if (interaction.getTypeId().equals(MqttPublish.TYPE_ID)) {
            receiveInteraction((MqttPublish) interaction);
        }
    }

    private void receiveInteraction(MqttPublish interaction) {
        if (pubDirect) {
            publish(interaction.getTopic(), interaction.getPayload(), interaction.getQos(), interaction.isRetained());
        } else {
            queue.add(interaction);
        }
    }


    public void store(String topic, String id,  AbstractMqttMessage message) {
        if (strategyMap.containsKey(topic)) {
            strategyMap.get(topic).store(topic, id, message, new StoreParameter.Builder()
                    .federateId(getId()).rti(rti).build());
        }
    }

    public void publish(String topic, Object object, int qos, boolean retained) {
        String message = JsonUtils.serialize(object);
        try {
            pub.publish(topic, message.getBytes(StandardCharsets.UTF_8), qos, retained);
        } catch (MqttException e) {
            throw new RuntimeException("publish to mqtt error, topic "+ topic + ", message " + message, e);
        }
    }

    public void subscribe(String topic) {
        try {
            sub.subscribe(topic);
        } catch (MqttException e) {
            throw new RuntimeException("subscribe mqtt topic " +  topic, e);
        }
    }

    @Override
    public boolean isTimeConstrained() {
        return false;
    }

    @Override
    public boolean isTimeRegulating() {
        return false;
    }


    public enum StoreStrategy {

        SIMPLE {
            @Override
            public void store(String topic, String id,  AbstractMqttMessage message, StoreParameter parameter) {
               // nop
            }
        },
        VEHICLE_MOVE {
            @Override
            public void store(String topic, String id,  AbstractMqttMessage message, StoreParameter parameter) {
                ObuProtocol payload = ((ObuMessage) message).getPayload();

                RtiAmbassador rti = parameter.getRti();
                long simulationTime = rti.getSimulationTime();

                // need to create a vehicle in simulation world
                String name = vehicleIdMap.getOrDefault(id, NameGenerator.getVehicleName());

                /*Boolean existYet = false;

                try {
                    existYet = (Boolean) rti.delegate("sumo", "existExternalVehicle",
                            new Class<?>[]{String.class}, new Object[]{name});
                } catch (Exception ex) {
                    throw new RuntimeException("Could not update location.", ex);
                }*/

                if (simulationTime == 0) {
                    // wait to create vehicle
                    return;
                }


                if (!vehicleIdMap.containsKey(id)) {

                    vehicleIdMap.put(id, name);

                    CVehicleAdd vehicleMove = mqtt.getVehicleMove();

                    VehicleFederateAssignment vehicleFederateAssignment = new VehicleFederateAssignment(simulationTime,
                            name, parameter.getFederateId(), 0D, vehicleMove.getProtoTypeName(),
                            vehicleMove.extractVehicleDeparture(), vehicleMove.getApplications());

                    try {
                        rti.triggerInteraction(vehicleFederateAssignment);
                    } catch (IllegalValueException | InternalFederateException ex) {
                        throw new RuntimeException("Could not send interaction to RTI.", ex);
                    }
                }

                List<VehicleData> vehicleUpdateData = new ArrayList<>();
                VehicleUpdates vehicleUpdates = new VehicleUpdates(simulationTime, new ArrayList<>(), vehicleUpdateData,
                        new ArrayList<>());
                ObuProtocol.Position p = payload.getBody().getPosition();;
                VehicleData data = new VehicleData.Builder(simulationTime, name)
                        .position(new MutableGeoPoint(p.getLatitude(), p.getLongitude()), null)
                        .orientation(null, payload.getBody().getHeading(), 0D).create();
                vehicleUpdateData.add(data);

                try {
                    rti.delegate("sumo", "setLastPositions", new Class<?>[]{vehicleUpdates.getClass()},
                            new Object[]{vehicleUpdates});
                } catch (Exception ex) {
                    throw new RuntimeException("Could not update location.", ex);
                }
            }
        };

        public void store(String topic, String id, AbstractMqttMessage message, StoreParameter parameter) {
            throw new RuntimeException("don not support store message.");
        }
    }


    static {
        strategyMap.put(MqttTopic.SIMPLE.getTopic(), StoreStrategy.SIMPLE);
        strategyMap.put(MqttTopic.VEHICLE_MOVE.getTopic(), StoreStrategy.VEHICLE_MOVE);
    }


    public class MqttProcess implements Runnable {

        @Override
        public void run() {
            while (true) {
                try {
                    MqttPublish interaction = queue.poll();
                    if (interaction != null) {
                        publish(interaction.getTopic(), interaction.getPayload(), interaction.getQos(), interaction.isRetained());
                    }
                } catch (Exception e) {
                    log.warn("Mqtt pub task fail", e);
                }
            }
        }
    }
}
